package com.relytosoft.mizar.analyze.system.services.consumer;

import com.alibaba.fastjson.JSON;
import com.relytosoft.mizar.analyze.base.constant.CommonKafkaConstants;
import com.relytosoft.mizar.analyze.system.api.entity.AyAlarm;
import com.relytosoft.mizar.analyze.system.services.service.AyAlarmService;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.stream.Collectors;

import static com.relytosoft.mizar.analyze.base.constant.MediaMqConstants.UNIFY_ALARM_TOPIC;

/**
 * @author wss
 * @version 1.0
 * @date 2022/5/24 19:05
 */
@Component
public class AyAlarmConsumer {

    protected Logger log = LoggerFactory.getLogger(this.getClass());

    @Autowired
    private AyAlarmService ayAlarmService;

    @KafkaListener(topics = UNIFY_ALARM_TOPIC, containerFactory = CommonKafkaConstants.BATCH_CONTAINER_FACTORY)
    public void handleAlarm(List<ConsumerRecord<?, ?>> records, Acknowledgment ack) {
        try {
            List<AyAlarm> alarmList = records
                    .stream()
                    .map(r -> JSON.parseObject(r.value().toString(), AyAlarm.class))
                    .collect(Collectors.toList());
            ayAlarmService.saveBatch(alarmList);
        } catch (Exception e) {
            log.info("报警数据处理异常---{}", records, e);
        }
        ack.acknowledge();
    }
}
